package com.bw.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.been.OrderInfo;
import com.bw.been.OrderWide;
import com.bw.been.PaymentInfo;
import com.bw.func.AsyncDimFunction;
import com.bw.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.concurrent.TimeUnit;

public class InfoWide {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        String groupid="dwm_Info_Wide";
        String orderInfotopic="dwd_order_info";
        String paymentInfotopic="dwd_payment_info";

        FlinkKafkaConsumer<String> OrderSource = MyKafkaUtil.getKafkaSource(groupid, orderInfotopic);
        FlinkKafkaConsumer<String> Paymentsource = MyKafkaUtil.getKafkaSource(groupid, paymentInfotopic);
        DataStreamSource<String> OrderSC = env.addSource(OrderSource);
        DataStreamSource<String> PaymentSC = env.addSource(Paymentsource);

//        OrderSC.print("OrderInfo>>>>>");
//        PaymentSC.print("PaymentInfo>>>>>");

        SingleOutputStreamOperator<OrderInfo> OrderInfoDS = OrderSC.map(new MapFunction<String, OrderInfo>() {
            @Override
            public OrderInfo map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                OrderInfo data = JSONObject.parseObject(jsonObject.getString("data"), OrderInfo.class);
                return data;
            }
        }).assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner((event, timestamp) -> {
                            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                            try {
                                return sdf.parse(event.getOrderTime()).getTime();
                            } catch (ParseException e) {
                                throw new RuntimeException(e);
                            }
                        })
        );
        OrderInfoDS.print("Order>>>>>>>>>>>>");

        SingleOutputStreamOperator<PaymentInfo> PaymentInfoDS = PaymentSC.map(new MapFunction<String, PaymentInfo>() {
            @Override
            public PaymentInfo map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                PaymentInfo data = JSONObject.parseObject(jsonObject.getString("data"), PaymentInfo.class);
                return data;
            }
        }).assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<PaymentInfo>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner((event, timestamp) -> Long.parseLong(event.getOrderTime()))
        );
        PaymentInfoDS.print("Pay>>>>>>>>>>>>");

        SingleOutputStreamOperator<OrderWide> order_wide = OrderInfoDS
                .keyBy(t -> t.getPolno())
                .intervalJoin(PaymentInfoDS.keyBy(t -> t.getPolno()))
                .between(Time.days(-2), Time.days(2))
                .process(new ProcessJoinFunction<OrderInfo, PaymentInfo, OrderWide>() {
                    @Override
                    public void processElement(OrderInfo orderInfo,
                                               PaymentInfo paymentInfo,
                                               ProcessJoinFunction<OrderInfo, PaymentInfo, OrderWide>.Context context,
                                               Collector<OrderWide> collector) throws Exception {
                        collector.collect(new OrderWide(orderInfo, paymentInfo));
                    }
                });

        order_wide.print("wide------------------------");

        String topic="dws_info_wide";

        SingleOutputStreamOperator<String> wide = order_wide.map(new MapFunction<OrderWide, String>() {
            @Override
            public String map(OrderWide orderWide) throws Exception {
                return JSONObject.toJSONString(orderWide);
            }
        });

        wide.addSink(MyKafkaUtil.getKafkaSink(topic));




        env.execute();
    }
}
